package com.example.mq1.mqserver.datacenter;

import com.example.mq1.common.MqException;
import com.example.mq1.mqserver.core.Binding;
import com.example.mq1.mqserver.core.Exchange;
import com.example.mq1.mqserver.core.MSGQueue;
import com.example.mq1.mqserver.core.Message;
import com.sun.rowset.internal.CachedRowSetWriter;
import org.apache.ibatis.annotations.Param;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 内存数据管理中心
 */
public class MemoryDataCenter {

    //交换机
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //队列
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    //绑定(第一个 key 是 exchangeName， 第二个 key 是 queueName)
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingMap = new ConcurrentHashMap<>();
    //消息
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //队列和消息的绑定关系
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //未确认的消息(第一个 key 是队列名，第二个 key 是 messageId)
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getExchangeName(), exchange);
        System.out.println("[MemoryDataCenter] 交换价添加成功！ exchangeName=" + exchange.getExchangeName());
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功！ exchangeName=" + exchangeName);
    }

    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getQueueName(), queue);
        System.out.println("[MemoryDataCenter 队列添加成功！ queueName=" + queue.getQueueName());
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter 队列删除成功！ queueName=" + queueName);
    }

    public void insertBinding(Binding binding) throws MqException {
//        ConcurrentHashMap<String, Binding> bindingMapNew = bindingMap.get(binding.getExchangeName());
//        if(bindingMapNew != null) {
//            throw new MqException("[MemoryDataCenter] 当前绑定已存在！ exchangeName=" + binding.getExchangeName() +
//                    "queueName=" + binding.getExchangeName());
//        }
//        bindingMapNew = new ConcurrentHashMap<>();
//        bindingMapNew.put(binding.getExchangeName(), binding);
//        bindingMap.put(binding.getExchangeName(), bindingMapNew);
//        System.out.println("[MemoryDataCenter] 绑定添加成功！exchangeName=" + binding.getExchangeName() +
//                ", queueName=" + binding.getExchangeName());
        //以上逻辑可以替换成以下更高级的玩法
        //不存在则创建一个新的 ConcurrentHashMap
        ConcurrentHashMap<String, Binding> bindingMapNest = bindingMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());
        synchronized (bindingMapNest) {
            if(bindingMapNest.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 当前绑定已存在！ exchangeName=" + binding.getExchangeName() +
                        ", queueName=" + binding.getQueueName());
            }
            bindingMapNest.put(binding.getQueueName(), binding);
        }
        System.out.println("[MemoryDataCenter] 绑定添加成功！ exchangeName=" + binding.getExchangeName() +
                ", queueName=" + binding.getQueueName());
    }

    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMapNest = bindingMap.get(exchangeName);
        //假定这里发生锁冲突的概论不大，因为没有必要再去创建一个锁对象
        if(bindingMapNest == null) {
            return null;
        }
        return bindingMapNest.get(queueName);
    }

    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingMap.get(exchangeName);
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMapNest = bindingMap.get(binding.getExchangeName());
        if(bindingMapNest == null) {
            throw new MqException("[MemoryDataCenter] 绑定不存在，删除失败！ exchangeName=" + binding.getExchangeName() +
                    ", queueName=" + binding.getQueueName());
        }
        bindingMapNest.remove(binding.getQueueName());
        System.out.println(("[MemoryDataCenter] 绑定删除成功！ exchangeName=" + binding.getExchangeName() +
                ", queueName=" + binding.getQueueName()));
    }

    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 消息添加成功！ messageId=" + message.getMessageId());
    }

    public Message getMessages(String messageId) {
        return messageMap.get(messageId);
    }

    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息删除成功！ messageId=" + messageId);
    }

    /**
     * 发送消息给指定队列
     * @param message
     */
    public void sendMessage(MSGQueue queue, Message message) throws MqException {
        LinkedList<Message> queueMessageList = queueMessageMap.computeIfAbsent(queue.getQueueName(),
                k -> new LinkedList<>());
        //这里加锁还有个原因时 LinkedList 是线程不安全的
        synchronized (queueMessageList) {
            queueMessageList.add(message);
        }
        //消息中也添加一份，即使相同的消息存在也没事(put操作直接覆盖)
        addMessage(message);
    }

    /**
     * 从对应队列中取出消息
     * 从 LinkeList 中头删
     * @param queueName
     * @return
     */
    public Message pollMessage(String queueName) {
        LinkedList<Message> queueMessageList = queueMessageMap.get(queueName);
        if(queueMessageList == null) {
            return null;
        }
        synchronized (queueMessageList) {
            if(queueMessageList.size() == 0) {
                return null;
            }
            Message message = queueMessageList.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出！ queueName=" + queueName +
                    ", messageId=" + message.getMessageId());
            return message;
        }
    }

    /**
     * 从指定队列中获取消息个数
     * @param queueName
     * @return
     */
    public int getMessageCount(String queueName) {
        LinkedList<Message> queueMessageList = queueMessageMap.get(queueName);
        if(queueMessageList == null) {
            return 0;
        }
        //要小心 LinkedList 是线程不安全的
        synchronized (queueMessageList) {
            return queueMessageList.size();
        }
    }


    /**
     * 添加未确认的消息
     * @param queueName
     * @param message
     */
    public void addMessageWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> messageWaitAckNest = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        messageWaitAckNest.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 添加未确认消息成功！ queueName=" + queueName +
                "messageId=" + message.getMessageId());
    }

    /**
     * 取出队列中指定的未确认消息
     * @param queueName
     * @param messageId
     */
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageWaitAckNest = queueMessageWaitAckMap.get(queueName);
        if(messageWaitAckNest == null) {
            return null;
        }
        return messageWaitAckNest.get(messageId);
    }

    public void removeMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageWaitAckNest = queueMessageWaitAckMap.get(queueName);
        if(messageWaitAckNest == null) {
            return;
        }
        messageWaitAckNest.remove(messageId);
        System.out.println("[MemoryDataCenter] 待确认消息以删除！ queueName=" + queueName +
                "messageId=" + messageId);
    }

    /**
     * 恢复硬盘数据
     * 读取硬盘数据到内存中
     */
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, ClassNotFoundException, MqException {
        //先清除内存上的所有数据(防止干扰)
        exchangeMap.clear();
        queueMap.clear();
        bindingMap.clear();
        messageMap.clear();
        queueMessageMap.clear();

        //恢复交换机数据
        List<Exchange> exchanges = diskDataCenter.selectExchangeAll();
        for(Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getExchangeName(), exchange);
        }
        //恢复队列数据
        List<MSGQueue> queues = diskDataCenter.selectQueueAll();
        for(MSGQueue queue : queues) {
            queueMap.put(queue.getQueueName(), queue);
        }
        //恢复绑定数据
        List<Binding> bindings = diskDataCenter.selectBindingAll();
        for(Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> bindingMapNest =  bindingMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingMapNest.put(binding.getQueueName(), binding);
        }
        //恢复消息数据
        for(MSGQueue queue : queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getQueueName());
            queueMessageMap.put(queue.getQueueName(), messages);
            for(Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }
        //未确认的消息这里可以不用管，因为硬盘中本身也没有相关设定
    }
}
